package plgmqtt

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgnet"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"sync"
	"time"
)

/*
不要订阅重复的主题

	比如/a/b/c, /a/+/c, 这样触发多次接收
*/
type PlgMqttExV2 struct {
	pubsyncflag int8 // 0：发送时是用队列， 1：同步发送(调用者同步)
	lk          sync.RWMutex
	workers     *utils.MultiGroupTask
	PlgBase

	firstStartup time.Time

	retained bool
	qos      byte

	msglst       *utils.SyncCycleList
	mqtt         *MqttInner
	sublst       map[string]*PlgMqttExSubItem
	monitor_recv *utils.MonitorSpeedSizeRec
}

type PlgMqttExSubItem struct {
	id       interface{}
	subtopic string
	cb       utils.SubscribeCallback
}

/*
*

			  {
			    "id": "xxx",
			    "type": "plg.mqtt.ex.v2",
			    "conf": {
			        "pubsyncflag":0,  // 默认0: 发送时使用队列发送, 1: 同步发送，是用发送调用者
		            "retained":false,
	                "qos":0,
					"clientid":"",
					"addr":"tcp://139.199.7.76:52883",
					"user":"",
					"pass":""
			    }
			  }
*/
func (this *PlgMqttExV2) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
		this.firstStartup = time.Now()
	}

	this.pubsyncflag = conf.Int8ByName("pubsyncflag", 0)
	this.qos = byte(conf.Int8ByName("qos", 0))
	this.retained = conf.BoolByName("retained", false)

	if this.workers == nil {
		this.workers = utils.NewMultiGroupTask()
		this.workers.ConfigWorkType(2)
		this.workers.ConfigMinWork(1)
		this.workers.ConfigMaxWork(5)
		this.workers.Start()
	}

	if this.monitor_recv == nil {
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.monitor_recv)
	}

	if this.sublst == nil {
		this.sublst = make(map[string]*PlgMqttExSubItem)
	}

	if this.mqtt == nil {
		this.mqtt = NewMqttInner()
	}
	this.mqtt.ReloadConf(conf)

	this.mqtt.OnMessage = func(sender *MqttInner, channelid string, data []byte) {
		this.monitor_recv.DeltaBuf(data)
		var isBreak bool = false
		for _, v := range this.sublst {
			if utils.RouteIncludesTopic_1(v.subtopic, channelid) {
				if v.cb != nil {
					v.cb(this, channelid, v.id, data, nil, &isBreak)
				}
				if isBreak {
					break
				}
			}
		}
	}

	utils.DefaultPatrolTask().AddTask(this, time.Second*30, func(id interface{}, args ...interface{}) {
		//utils.DefaultWorkers().PostTaskFunc(this, this.innerWork)
	})
	return nil
}

func (this *PlgMqttExV2) Close() (err error) {
	this.mqtt.Close()
	this.workers.Stop()
	utils.Monitor.RemoveMonitorRec(this.monitor_recv)
	this.PlgBase.Close()
	utils.DefaultPatrolTask().DelTask(this)
	return
}

func (this *PlgMqttExV2) innerPublish(topic string, data interface{}) error {
	var dispBuf []byte
	dispflag := 1
	if buf, ok := data.([]byte); ok {
		dispBuf = buf
	} else if json, ok := data.(*wrapper.SuperValue); ok {
		dispBuf = json.AsJSONBytesEx(false, false)
	} else if str, ok := data.(string); ok {
		dispBuf = []byte(str)
	} else {
		dispflag = 0
	}
	if dispflag == 1 {
		return this.mqtt.Publish0(topic, 0, this.retained, dispBuf)
	}
	return fmt.Errorf("未知的数据类型")
}

func (this *PlgMqttExV2) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	if this.pubsyncflag == 1 {
		this.innerPublish(channelid, data)
	} else {
		this.workers.PostTaskFunc(this.PlugId, func() {
			this.innerPublish(channelid, data)
		})
	}
	return
}

func (this *PlgMqttExV2) OnHttpStatus(rec *plgnet.PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()
		json.SetKeyValue("recv", this.monitor_recv.Info())
		json.SetKeyValues("pubsyncflag", this.pubsyncflag)
		json.SetKeyValue("qos", this.qos)
		json.SetKeyValue("retained", this.retained)
		json.SetKeyValue("sub-topics", this.mqtt.SubTopics(" "))
		json.SetKeyValue("mqtt-status", this.mqtt.StatusString())
		json.SetKeyValue("startup", utils.DateTimeString3(this.firstStartup))
		json.SetKeyValues("now", utils.NowString())
		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, true))
		} else {
			PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
		}
	} else if itype == 98 {
		rec.ResponseSyncListNEx(this.msglst)
	} else if itype == 1 {
		json := wrapper.NewSVArray()
		//searchval := rec.QryAsString("searchval", "")
		//sn := 0
		//maxn := rec.QryAsInt("n", 100)
		//this.lst.ListRangeV2(func(key, val interface{}) bool {
		//	pojo := val.(*VrsSitePojo)
		//	ok := true
		//	if len(searchval) > 0 {
		//		ok = strings.Contains(pojo.id, searchval)
		//	}
		//	return ok
		//}, func(k0, v0, k1, v1 interface{}) bool {
		//	p0 := v0.(*VrsSitePojo)
		//	p1 := v1.(*VrsSitePojo)
		//	return p0.id < p1.id
		//}, func(key, val interface{}) bool {
		//	sn++
		//	pojo := val.(*VrsSitePojo)
		//	itm := json.AppendObject()
		//	itm.SetKeyValue("sn", sn)
		//	itm.SetKeyValue("id", pojo.id)
		//	return sn < maxn
		//})

		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, false))
		} else {
			PrjHttpUtils.JSON2Table(rec.RespW, "", json.AsJSONString(false))
		}
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "lst", "status?type=1&n=100&searchval=")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}
	return
}

/*
忽略channelid
*/
func (this *PlgMqttExV2) SubscribeChannel(channelid string, id interface{}, cb utils.SubscribeCallback) {
	this.lk.Lock()
	defer this.lk.Unlock()
	itm := &PlgMqttExSubItem{id: id, subtopic: channelid, cb: cb}
	keystr := fmt.Sprintf("%p-%s", id, channelid)
	this.mqtt.Subscribe(channelid)
	this.sublst[keystr] = itm
}

func (this *PlgMqttExV2) UnsubscribeChannel(channelid string, id interface{}) bool {
	this.lk.Lock()
	defer this.lk.Unlock()
	keystr := fmt.Sprintf("%p-%s", id, channelid)
	delete(this.sublst, keystr)
	flag := 0
	for _, v := range this.sublst {
		if channelid == v.subtopic {
			flag = 1
			break
		}
	}
	if flag == 0 {
		this.mqtt.Unsubscribe(channelid)
	}
	return true
}

func (this *PlgMqttExV2) LogInfoMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *PlgMqttExV2) LogWarnMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreatePlgMqttExV2Func(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgMqttExV2{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
